package com.esdk.utils;

import com.esdk.esdk;
import com.esdk.interfaces.Callback;

import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author ffychina@163.com
 * @since 2025-9-14
 * 定时批量执行器，可在间隔时间内或数量满足条件时批量执行
 * */
public class IntervalExcutor<T>{
  private Thread intervalThread = null;
  private boolean stopped=false;
  private long lastTime=0;
  private int queueMaxSize=100; //达到或超过队列上限就立即处理
  private int queueMinSize=0; //小于或等于队列下限就立即处理，这是为了空闲时能立即响应，为0则不处理
  private Callback<List<T>> tasks;
  private ConcurrentLinkedQueue<T> queue=new ConcurrentLinkedQueue<>();
  private boolean isOtherThreadRunning=false;

  public IntervalExcutor(long sleepMs,int queueLimit,Callback<List<T>> r){
    init(sleepMs,queueLimit,"interval-thread-"+Thread.currentThread(),r);
  }

  public IntervalExcutor(long sleepMs,int queueLimit,String threadName,Callback<List<T>> r){
    init(sleepMs,queueLimit,threadName,r);
  }

  private void init(long sleepMs,int queueLimit,String threadName,Callback<List<T>> r){
    queueMaxSize=queueLimit;
    tasks=r;
    intervalThread=new Thread(()->{
      while(!stopped){
        long startTime=System.currentTimeMillis();
        if(!queue.isEmpty()&&!isOtherThreadRunning){
          isOtherThreadRunning=true;
          exec(tasks);
          isOtherThreadRunning=false;
        }
        lastTime=System.currentTimeMillis();
        long sleepTime=esdk.math.max(0,sleepMs-(lastTime-startTime));
//        esdk.sout("sleeptime:",sleepTime);
        esdk.tool.sleep(sleepTime);
      }
    });
    intervalThread.setName(threadName);
    intervalThread.start();
  }

  private void asyncExec(Callback<List<T>> r){
    ThreadUtils.async(()->{
      isOtherThreadRunning=true;
      exec(r);
      isOtherThreadRunning=false;
    });
  }


  private void exec(Callback<List<T>> r){
    synchronized(queue){
      List<T> ts=esdk.array.poll(queue,queueMaxSize);
      try{
        r.call(ts);
      }catch(Exception e){
        throw esdk.tool.wrapThrowble(e);
      }
    }
  }

  public void stop(){
    stopped=true;
  }

  public void add(T item){
    queue.add(item);
    if(!isOtherThreadRunning && queue.size()>=queueMaxSize){
      asyncExec(tasks);
    }else if(queue.size()<=queueMinSize){
      asyncExec(tasks);
    }
  }

  public IntervalExcutor<T> setQueueMaxSize(int queueMinSize){
    this.queueMaxSize=queueMaxSize;
    return this;
  }

  public IntervalExcutor<T> setQueueMinSize(int queueMinSize){
    this.queueMinSize=queueMinSize;
    return this;
  }

  public IntervalExcutor<T> setQueueSize(int queueMinSize,int queueMaxSize){
    return setQueueMinSize(queueMinSize).setQueueMaxSize(queueMaxSize);
  }
}
